class Observer {
  constructor() {
    this._opts = [];
    this._unsubscribeCallback = null;
  }

  next(data) {
    this._opts.forEach((opt) => {
      const { next: nextCb } = opt;
      nextCb && nextCb(data);
    });
  }

  addOpt(val) {
    this._opts.push(val);
  }

  unsubscribe() {
    this._opts = [];
    this._unsubscribeCallback && this._unsubscribeCallback();
  }
}

class Observable {
  constructor(callback) {
    this._lastObservableObj = null; // 支持取消订阅时执行整个监听链的回调函数
    this._dataGenerator = callback;
    this._observer = new Observer();
  }

  subscribe(opt) {
    this._observer.addOpt(opt);
    this._observer._unsubscribeCallback = this._dataGenerator(this._observer);
  }

  unsubscribe() {
    this._observer.unsubscribe();
    // 执行监听链上上一个Observable的回调
    if(this._lastObservableObj) this._lastObservableObj.unsubscribe();
  }

  /**
   * 返回Observable总处理函数
   * 每个作为pipe参数的函数，都是一个Observable实例的处理函数
   */
  _getObservableHandler(fns) {
    if (fns.length === 0) return () => this;
    return (obj) => {
      // prev: 处理后的obj
      return fns.reduce((prev, fn) => {
        return fn(prev);
      }, obj);
    };
  }

  pipe(...fns) {
    return this._getObservableHandler(fns)(this);
  }
}

function map(cb) {
  return (preObservableObj) => {
    const newObservableObj = new Observable((observer) => {
      preObservableObj.subscribe({
        next(value) {
          observer.next(cb(value))
        }
      })
    });
    newObservableObj._lastObservableObj = preObservableObj;
    return newObservableObj
  };
}

function bufferCount(num) {
  return (preObservableObj) => {
    const newObservableObj = new Observable((observer) => {
      const bufferArr = [];
      preObservableObj.subscribe({
        next(value) {
          bufferArr.push(value);
          observer.next(bufferArr.slice(-num))
        }
      })
    });
    newObservableObj._lastObservableObj = preObservableObj;
    return newObservableObj
  };
}

const source = new Observable((observer) => {
  console.log("已执行callback");
  setInterval(() => {
    observer.next(1);
  }, 1000);

  return () => {
    console.log("取消订阅");
  };
}).pipe(
  map((i) => i + 1),
  map((i) => i * 2),
  bufferCount(3)
);

setTimeout(() => {
  source.subscribe({
    next: (data) => {
      console.log(data);
    },
  });
}, 2000);

setTimeout(() => {
  source.unsubscribe();
}, 7000);
